/**
 * Copyright (C) 2012-2013 jumei, Inc.
 * RpcInvorkHandler.java
 * 2015年6月25日
 */
package org.goface.rpc.io;

import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import org.goface.rpc.Message;
import org.goface.rpc.MessageContext;
import org.goface.rpc.pipeline.ServicePipeLine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.util.concurrent.SettableFuture;

/**
 * @author jm110
 */
public class RpcClientSideHandler extends ChannelHandlerAdapter {
    private Logger LOG = LoggerFactory.getLogger(RpcClientSideHandler.class);
    private ServicePipeLine servicePipeLine;

    public RpcClientSideHandler(ServicePipeLine servicePipeLine) {
        this.servicePipeLine = servicePipeLine;
    }

    private Cache<String, MessageContext> cache = CacheBuilder.newBuilder().maximumSize(10000)
            .expireAfterWrite(1, TimeUnit.MINUTES).build();

    private MessageContext getWaitingFuture(String id) {
        MessageContext context = null;
        try {
            context = cache.get(id, new Callable<MessageContext>() {
                @Override
                public MessageContext call() throws Exception {
                    return null;
                }
            });
            cache.invalidate(id);

        } catch (ExecutionException e) {
            LOG.error("failed to get futrue in cache", e);
        }
        return context;

    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            Message responseMessage = Message.class.cast(msg);
            MessageContext context = getWaitingFuture(responseMessage.getId());
            if (context == null) {
                LOG.warn("receive a timeout response message,dropped it.");

            } else {
                context.getProperties().put(MessageContext.MESSAGE_TYPE, MessageContext.MESSAGE_TYPE_REPLY);

                context.setMessage(responseMessage);
                if (responseMessage.getCause() != null) {
                    // 服务调用异常
                    servicePipeLine.fireException(new RuntimeException(responseMessage.getCause()));
                } else {
                    servicePipeLine.fireReceiveResponse(context);
                }
            }
        } catch (Throwable th) {
            LOG.error("exception in receive response", th);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 无法确定是哪一次调用的异常，直接记录日志，上层调用自己超时结束
        LOG.error("exception caught pls check.", cause);
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object context, ChannelPromise promise) throws Exception {
        // SettableFuture<Message> future = SettableFuture.create();
        MessageContext request = MessageContext.class.cast(context);
        Message msg = request.getMessage();
        cache.put(msg.getId(), request);
        ctx.writeAndFlush(msg, promise);

    }

}
